// 异步工作器

#ifndef __M_LOPPER_H__
#define __M_LOPPER_H__

#include "buffer.hpp"
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

namespace gjclog
{
    using Functor = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全模式：缓冲区满了就阻塞，避免资源耗尽
        ASYNC_UNSAFE // 非安全模式：缓冲区满了就扩容，用于性能测试
    };

    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb, AsyncType lopper_type = AsyncType::ASYNC_SAFE)
            : _stop(false), _thread(std::thread(&AsyncLooper::threadEntry, this)), _callBack(cb), _lopper_type(lopper_type)
        {
        }
        ~AsyncLooper()
        {
            stop();
        }
        void stop()
        {
            _stop = true;           // 设置退出标志
            _cond_con.notify_all(); // 唤醒所有工作线程
            _thread.join();         // 等待工作线程退出
        }
        void push(const char *data, size_t len)
        {
            // 1. 无限扩容    2. 固定大小，生产缓冲区满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，缓冲区剩余空间大于数据长度就可以添加数据
            if (_lopper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; });
            // 满足条件即可向缓冲区添加数据
            _pro_buf.push(data, len);
            // 唤醒消费者处理缓冲区数据
            _cond_con.notify_one();
        }

    private:
        // 线程入口函数
        void threadEntry()
        {
            while (1)
            {
                {
                    // 1. 判断生产缓冲区有没有数据，有就交换，没有不交换
                    std::unique_lock<std::mutex> lock(_mutex);
                    
                    // 若退出标志设置并且生产缓冲区无数据，则可以退出循环
                    if(_stop && _pro_buf.empty()) break;

                    _cond_con.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    // 2. 唤醒生产者
                    if (_lopper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 3. 对消费者缓冲区进行处理
                _callBack(_con_buf);
                // 4. 初始化重置消费换缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 数据处理的回调函数，由使用者传入

    private:
        AsyncType _lopper_type;  // 缓冲区状态：满了以后（扩容/阻塞）
        std::atomic<bool> _stop; // 工作器是否停止
        Buffer _pro_buf;         // 生产缓冲区
        Buffer _con_buf;         // 消费缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro; // 生产条件变量
        std::condition_variable _cond_con; // 消费条件变量
        std::thread _thread;               // 异步工作器工作线程
    };
}

#endif